package com.deng.eventflow.core.tmf;

import com.deng.eventflow.core.Connector;
import com.deng.eventflow.core.Processor;
import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.functions.Function;

import java.util.HashMap;
import java.util.Map;
import java.util.Queue;

public class Slice extends AbstractSlice {
    /**
     * The last value output by the processor for each slice
     */
    protected HashMap<Object, Object> m_lastValues;

    /**
     * Creates a dummy slice processor. This constructor is only used for
     * deserialization purposes.
     */
    protected Slice()
    {
        super();
    }

    /**
     * Creates a new Slice processor.
     *
     * @param func The slicing function
     * @param proc The processor to apply on each slice
     * @param clean_func The cleaning function
     */
    public Slice(/*@ non_null @*/ Function func, /*@ non_null @*/ Processor proc,
                                  Function clean_func)
    {
        super(func, proc, clean_func);
        m_lastValues = new HashMap<Object, Object>();
    }

    /**
     * Creates a new Slice processor.
     * @param func The slicing function
     * @param proc The processor to apply on each slice
     */
    public Slice(/*@ non_null @*/ Function func, /*@ non_null @*/ Processor proc)
    {
        this(func, proc, null);
    }

    /**
     * @since 0.10.2
     */
    @Override
    public Object printState()
    {
        Map<String,Object> contents = new HashMap<String,Object>();
        contents.put("cleaning-function", m_cleaningFunction);
        contents.put("explode-arrays", m_explodeArrays);
        contents.put("last-values", m_lastValues);
        contents.put("sinks", m_sinks);
        contents.put("slices", m_slices);
        contents.put("processor", m_processor);
        contents.put("slicing-function", m_slicingFunction);
        return contents;
    }

    /**
     * @since 0.10.2
     */
    @SuppressWarnings("unchecked")
    @Override
    public Slice readState(Object o) throws ProcessorException
    {
        Map<String,Object> contents = (HashMap<String,Object>) o;
        Function cleaning_function = (Function) contents.get("cleaning-function");
        Function slicing_function = (Function) contents.get("slicing-function");
        Processor proc = (Processor) contents.get("processor");
        Slice s = new Slice(slicing_function, proc, cleaning_function);
        s.m_lastValues = (HashMap<Object,Object>) contents.get("last-values");
        s.m_sinks = (HashMap<Object,QueueSink>) contents.get("sinks");
        s.m_slices = (HashMap<Object,Processor>) contents.get("slices");
        s.m_explodeArrays = (Boolean) contents.get("explode-arrays");
        // Connect slices to sinks
        for (Map.Entry<Object,Processor> entry : s.m_slices.entrySet())
        {
            Processor p = entry.getValue();
            if (!s.m_sinks.containsKey(entry.getKey()))
            {
                throw new ProcessorException("Cannot restore the state of the slice processor");
            }
            QueueSink qs = s.m_sinks.get(entry.getKey());
            Connector.connect(p, qs);
        }
        return s;
    }

    @Override
    public Slice duplicate(boolean with_state)
    {
        Slice s = null;
        if (m_cleaningFunction == null)
        {
            s = new Slice(m_slicingFunction.duplicate(with_state), m_processor.duplicate(with_state));
        }
        else
        {
            s = new Slice(m_slicingFunction.duplicate(with_state), m_processor.duplicate(with_state),
                    m_cleaningFunction.duplicate(with_state));
        }
        super.copyInto(s, with_state);
        if (with_state)
        {
            s.m_lastValues.putAll(m_lastValues);
        }
        return s;
    }

    @Override
    public void reset()
    {
        super.reset();
        m_lastValues.clear();
    }

    @Override
    protected boolean produceReturn(Queue<Object[]> outputs)
    {
        outputs.add(new Object[] {m_lastValues});
        m_outputCount++;
        return true;
    }

    @Override
    protected void handleNewSliceValue(Object slice_id, Object value, Queue<Object[]> outputs)
    {
        if (value != null)
        {
            m_lastValues.put(slice_id, value);
        }
    }

}
